package cn.jly.flink.watermark;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.time.Duration;

/**
 * watermark的本质
 * <p>
 * 水位线的出现是为了解决实时计算中的数据乱序问题，它的本质是 DataStream 中一个带有时间戳的元素。如果 Flink 系统中出现了一个 WaterMark T，
 * 那么就意味着 EventTime < T 的数据都已经到达，窗口的结束时间和 T 相同的那个窗口被触发进行计算了。
 * <p>
 * 也就是说：水位线是 Flink 判断迟到数据的标准，同时也是窗口触发的标记。
 * <p>
 * 在程序并行度大于 1 的情况下，会有多个流产生水位线和窗口，这时候 Flink 会选取时间戳最小的水位线。
 * <p>
 * 案例：
 * 模拟一个实时接收 Socket 的 DataStream 程序，代码中使用 AssignerWithPeriodicWatermarks 来设置水印，将接收到的数据进行转换，
 * 分组并且在一个 5秒的窗口内获取该窗口中第二个元素最小的那条数据。
 * <p>
 * 说明
 * 在这里要特别说明，Flink 在用时间 + 窗口 + 水印来解决实际生产中的数据乱序问题，有如下的触发条件：
 * <p>
 * - watermark 时间 >= window_end_time；
 * <p>
 * - 在 [window_start_time,window_end_time) 中有数据存在，这个窗口是左闭右开的。
 * <p>
 * 此外，因为 WaterMark 的生成是以对象的形式发送到下游，同样会消耗内存，因此水印的生成时间和频率都要进行严格控制，否则会影响我们的正常作业
 *
 * @author lanyangji
 * @create 2020-09-02 17:12
 */
@Slf4j
public class WatermarkApp {
    public static void main(String[] args) throws Exception {
        // 参数
        ParameterTool parameterTool = ParameterTool.fromArgs(args);
        String host = parameterTool.get("host", "192.168.88.101");
        String port = parameterTool.get("port", "9999");

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 设置时间语义
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        // 设置水印生成的时间间隔
        env.getConfig().setAutoWatermarkInterval(100);

        env.socketTextStream(host, Integer.parseInt(port))
                .flatMap(new FlatMapFunction<String, MyData>() {
                    @Override
                    public void flatMap(String value, Collector<MyData> out) throws Exception {
                        String[] split = value.split(",");
                        if (split.length == 2) {
                            MyData myData = new MyData(split[0].trim(), Long.parseLong(split[1].trim()));
                            log.info("myData: {}", myData);
                            out.collect(myData);
                        }
                    }
                })
                .assignTimestampsAndWatermarks(
                        // 指定延时处理的时间，或者说允许数据迟到的时间3s
                        WatermarkStrategy.<MyData>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                                .withTimestampAssigner(new SerializableTimestampAssigner<MyData>() {
                                    @Override
                                    public long extractTimestamp(MyData element, long recordTimestamp) {
                                        log.info("eventTime: {}", element.getTimestamp());
                                        // 指定eventTime的时间字段
                                        return element.getTimestamp();
                                    }
                                })
                )
                .keyBy(new KeySelector<MyData, String>() {
                    @Override
                    public String getKey(MyData value) throws Exception {
                        return value.getName();
                    }
                })
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .minBy("timestamp")
                .printToErr();

        env.execute("WatermarkApp");
    }

    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class MyData {
        private String name;
        private Long timestamp;
    }
}
